package cn.baiyang.rocketmq.spring.autoconfigure;

import java.time.Duration;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.java.impl.ClientImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;
import org.springframework.util.Assert;

/**
 *
 * @author hongzhu
 * @version V1.0
 * @since 2023-07-24 17:32
 */
@Configuration
@EnableConfigurationProperties(RocketMQProperties.class)
@ConditionalOnClass({ ClientImpl.class})
@ConditionalOnProperty(prefix = "rocketmq-java", value = "endpoints")
@Import({ListenerContainerConfiguration.class, RocketMQListenerConfiguration.class})
public class RocketMQAutoConfiguration implements ApplicationContextAware {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQAutoConfiguration.class);

    public static final String PRODUCER_BEAN_NAME = "defaultProducer";

    @Autowired
    private Environment environment;

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @PostConstruct
    public void checkProperties() {
        String endpoints = environment.getProperty("rocketmq-java.endpoints", String.class);
        LOGGER.debug("rocketmq-java.endpoints = {}", endpoints);
        if (endpoints == null) {
            LOGGER.warn("The necessary spring property 'rocketmq-java.endpoints' is not defined, all rocketmq beans creation are skipped!");
        }
    }

    @Bean(PRODUCER_BEAN_NAME)
    @ConditionalOnMissingBean(Producer.class)
    @ConditionalOnProperty(prefix = "rocketmq-java", value = {"endpoints", "producer.topics"})
    public Producer defaultProducer(RocketMQProperties properties) throws ClientException {
        String endpoints = properties.getEndpoints();
        RocketMQProperties.Producer producerConfig = properties.getProducer();
        String topics = producerConfig.getTopics();
        Assert.hasText(endpoints, "[rocketmq-java.endpoints] must not be null");
        Assert.hasText(topics, "[rocketmq-java.producer.topics] must not be null");

        SessionCredentialsProvider sessionCredentialsProvider =
            new StaticSessionCredentialsProvider(properties.getAccesskey(), properties.getSecretkey());
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .setRequestTimeout(Duration.ofSeconds(properties.getRequestTimeout()))
            .enableSsl(properties.getEnableSsl())
            .setCredentialProvider(sessionCredentialsProvider)
            .build();
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        Producer producer = provider.newProducerBuilder()
            .setClientConfiguration(clientConfiguration)
            /*
             * 强烈推荐预设置应用内用到的所有topic，以便发现潜在问题
             */.setTopics(topics.split(","))
            // 设置重试次数，默认为3
            .setMaxAttempts(producerConfig.getMaxAttempts())
            .build();
        LOGGER.info(String.format("%s started successful on endpoints %s",  PRODUCER_BEAN_NAME, endpoints));
        return producer;
    }
}
